package com.shujia.transformation

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util
import scala.collection.JavaConversions._


object Demo09Split {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // 通过socket构建无界流
    val socketDS: DataStream[String] = env.socketTextStream("master", 8888)

    //    val splitStream: SplitStream[String] = socketDS
    //      .split(line => {
    //        val arrList: util.ArrayList[String] = new util.ArrayList[String]()
    //        if (line.toInt % 2 == 0) {
    //          arrList.add("even")
    //        } else {
    //          arrList.add("odd")
    //        }
    //        arrList.toList
    //      })
    //
    //    splitStream.select("even").print()

    val evenTag: OutputTag[String] = new OutputTag[String]("even")
    val oddTag: OutputTag[String] = new OutputTag[String]("odd")

    val outputStream: DataStream[String] = socketDS
      .process(new ProcessFunction[String, String] {
        override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
          if (value.toInt < 0) {
            out.collect(value)
          }
          else if (value.toInt % 2 == 0) {
            ctx.output(evenTag, value)
          } else {
            ctx.output(oddTag, value)
          }
        }
      })
    outputStream.getSideOutput(evenTag).print()


    env.execute()
  }

}
